package com.atguigu.chapter07;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/6 9:12
 */
public class Flink03_WindowFunction_Incre {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        // 切分
                        String[] line = value.split(",");
                        return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

                    }
                });


        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());

        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorKS
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        // TODO 窗口的 增量聚合 函数
        sensorWS
//                .reduce(new ReduceFunction<WaterSensor>() {
//                    @Override
//                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
//                        System.out.println(value1 + " <---------> " + value2);
//                        return new WaterSensor(value1.getId(), System.currentTimeMillis(), value1.getVc() + value2.getVc());
//                    }
//                })
                .aggregate(new AggregateFunction<WaterSensor, Long, String>() {
                    /**
                     * 初始化累加器
                     * @return
                     */
                    @Override
                    public Long createAccumulator() {
                        System.out.println("create Acc....");
                        return 0L;
                    }

                    /**
                     * 聚合逻辑：来一条数据，怎么处理
                     * @param value
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Long add(WaterSensor value, Long accumulator) {
                        System.out.println("add ...");
                        return accumulator + value.getVc();
                    }

                    /**
                     * 窗口输出的时候，调用最终的结果
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public String getResult(Long accumulator) {
                        System.out.println("result ...");
                        return accumulator.toString();
                    }

                    /**
                     * 只有在 会话窗口 才会调用到
                     * @param a
                     * @param b
                     * @return
                     */
                    @Override
                    public Long merge(Long a, Long b) {
                        System.out.println("调用了 merge方法 ...");
                        return a+b;
                    }
                })
                .print();


        env.execute();
    }
}
/*
    窗口的增量聚合函数：
        1）来一条数据，聚合一条
        2）在窗口结束的时候，输出一次

    reduce： 同一窗口内，同一分组的第一条数据来的时候，不会进入reduce方法
    aggregate: 相比reduce更加灵活，输入类型、中间状态、输出结果 都可以不一样
 */